package com.zhangsr.zk.client.listener;

import com.zhangsr.zk.client.listener.ZkEventThread.ZkEvent;
import com.zhangsr.zk.client.serialize.ZkClientSerializer;
import org.apache.log4j.Logger;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.springframework.util.CollectionUtils;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @Author: zhangsr
 */
public abstract class ZkClientListener extends ZkClientSerializer {

    private final static Logger log = Logger.getLogger(ZkClientListener.class);

    protected ZkEventThread eventThread;
    private KeeperState currentState;

    protected Map<String, Set<IZkDataListener>> dataListeners = new ConcurrentHashMap<String, Set<IZkDataListener>>();
    protected Map<String, Set<IZkChildListener>> childListeners = new ConcurrentHashMap<String, Set<IZkChildListener>>();
    private final Set<IZkConnectionListener> connectionListeners = new CopyOnWriteArraySet<IZkConnectionListener>();

    protected void initZkEventThread() {
        eventThread = new ZkEventThread();
        eventThread.start();
    }

    //------------------------ public --------------------------

    public void subscribeDataChanges(String path, IZkDataListener listener) {
        synchronized (this.dataListeners) {
            Set<IZkDataListener> listeners = this.dataListeners.get(path);
            if (listeners == null) {
                listeners = new CopyOnWriteArraySet<IZkDataListener>();
                this.dataListeners.put(path, listeners);
            }
            listeners.add(listener);
        }
    }

    public void unsubscribeDataChanges(String path, IZkDataListener dataListener) {
        synchronized (this.dataListeners) {
            final Set<IZkDataListener> listeners = this.dataListeners.get(path);
            if (listeners != null) {
                listeners.remove(dataListener);
            }
            if (listeners == null || listeners.isEmpty()) {
                this.dataListeners.remove(path);
            }
        }
    }

    public void subscribeConnectionChanges(final IZkConnectionListener listener) {
        synchronized (this.connectionListeners) {
            this.connectionListeners.add(listener);
        }
    }

    public void unsubscribeConnectionChanges(IZkConnectionListener stateListener) {
        synchronized (this.connectionListeners) {
            this.connectionListeners.remove(stateListener);
        }
    }
    //------------------------ protected --------------------------

    protected void processConnectionChanged(WatchedEvent event) {
        log.info("zookeeper state changed (" + event.getState() + ")");
        setCurrentState(event.getState());

        fireConnectionChangedEvent(event.getState());
        if (event.getState() == KeeperState.Expired) {
            try {
                reconnect();
                fireReConnectionEvent();
            } catch (final Exception e) {
                log.info("Unable to re-establish connection. Notifying consumer of the following exception: ", e);
                fireConnectionErrorEvent(e);
            }
        }
    }

    protected void processDataOrChildChange(WatchedEvent event) {
        final String path = event.getPath();

        if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged
                || event.getType() == Watcher.Event.EventType.NodeCreated
                || event.getType() == Watcher.Event.EventType.NodeDeleted) {
            Set<IZkChildListener> childListeners = this.childListeners.get(path);
            if (childListeners != null && !childListeners.isEmpty()) {
                fireChildChangedEvents(path, childListeners);
            }
        }

        if (event.getType() == Watcher.Event.EventType.NodeDataChanged
                || event.getType() == Watcher.Event.EventType.NodeDeleted
                || event.getType() == Watcher.Event.EventType.NodeCreated) {
            Set<IZkDataListener> listeners = this.dataListeners.get(path);
            if (listeners != null && !listeners.isEmpty()) {
                fireDataChangedEvents(event.getPath(), listeners);
            }
        }
    }

    /**
     * 判断节点上是否有监听
     *
     * @param path
     * @return
     */
    protected boolean hasListeners(String path) {
        Set<IZkDataListener> dataListeners = this.dataListeners.get(path);
        if (CollectionUtils.isEmpty(dataListeners)) {
            return true;
        }
        return false;
    }

    //------------------------ private --------------------------

    private void fireConnectionChangedEvent(final KeeperState state) {
        for (final IZkConnectionListener connectionListener : this.connectionListeners) {
            eventThread.send(new ZkEvent("State changed to " + state + " sent to " + connectionListener) {
                @Override
                public void run() throws Exception {
                    connectionListener.handleConnectionChanged(state);
                }
            });
        }
    }

    private void fireReConnectionEvent() {
        for (final IZkConnectionListener connectionListener : this.connectionListeners) {
            eventThread.send(new ZkEvent("New session event sent to " + connectionListener) {
                @Override
                public void run() throws Exception {
                    connectionListener.handleReConnection();
                }
            });
        }
    }

    private void fireConnectionErrorEvent(final Throwable error) {
        for (final IZkConnectionListener connectionListener : this.connectionListeners) {
            eventThread.send(new ZkEvent("Session establishment error(" + error + ") sent to " + connectionListener) {
                @Override
                public void run() throws Exception {
                    connectionListener.handleConnectionError(error);
                }
            });
        }
    }

    private void fireChildChangedEvents(final String path, Set<IZkChildListener> listeners) {
    }

    private void fireDataChangedEvents(final String path, Set<IZkDataListener> listeners) {
        for (final IZkDataListener listener : listeners) {
            eventThread.send(new ZkEvent("Data of " + path + " changed sent to " + listener) {
                @Override
                public void run() throws Exception {
                    // reinstall watch
                    exists(path, true);
                    try {
                        Object data = readData(path, null, true);
                        listener.handleDataChange(path, data);
                    } catch (Exception e) {
                        listener.handleDataDeleted(path);
                    }
                }
            });
        }
    }

    private void setCurrentState(KeeperState currentState) {
        this.currentState = currentState;
    }
}
